/* Copyright (C) 2008 Sahid Ferdjaoui <sahid@funraill.org>
 *
 *   This file is part of Postmemd.
 *
 * Postmemd is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */


#include "pm_core.h"
#include "pm_config.h"
#include "pm_log.h"
#include "pm_slaves.h"
#include "pm_poll.h"
#include "pm_table.h"

#include "ev.h"
#include "json.h"

#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>

#include <stdio.h>


extern config conf;

/** slaves pool */
pool_slaves_t * pool;


void
pm_core_process_loop (core_t * core)
{
	int closed = 0, result;

	while (!closed)
		{
			switch (core->state)
				{
				case pm_close:
					fprintf (stdout, "closed\n");
					pm_core_close (core);
					closed = 1;
					break;

				case pm_read_client:
					result = pm_poll_read_fd (core->sfd, & core->rbuffer, & core->rbuffer_bytes);
					switch (result)
						{
						case  0: core->state = pm_parse_json; break;
						case  1: core->state = pm_read_client; break;
						case -1: core->state = pm_close; break;
						}
					break;
				case pm_parse_json:
					if ((pm_core_parse_json (core->rbuffer, & core) == -1) ||
							(pm_core_json_cmd (core->json_obj, & core) == -1))
						{
							core->state = pm_close;
						}
					core->state = pm_dispatch;
					break;

				case pm_dispatch:
					switch (core->cmd)
						{
						case PM_CMD_ADD:
							/** record key and current slave in table */
							if (pm_table_get_pid (core->key) == -1)
								{
									fprintf (stdout, "enresitre dans la table\n");
									pm_table_add_pid (core->slave->pid, core->key);
									core->writed = 0;
									core->state = pm_write_slave;
									break;
								}
							fprintf (stdout, "deja dans la table\n");
							core->state = pm_close;
							break;
						}
					break;


				case pm_write_slave:
					result = pm_poll_write_fd (core->slave->write, core->rbuffer, & core->rbuffer_bytes, & core->writed);
					switch (result)
						{
						case  0: core->state = pm_read_slave; 
							//pm_record_fd (core->loop, & core->event, core->slave->read, pm_event_handler, EV_READ);
							free (core->rbuffer);
							core->rbuffer = NULL;
							sleep (2);
							return;
						case  1: core->state = pm_write_slave; break;
						case -1: core->state = pm_close; break;
						}
					break;


				case pm_read_slave:

					result = pm_poll_read_fd (core->slave->read, & core->rbuffer, & core->rbuffer_bytes);
					switch (result)
						{
						case  0: core->state = pm_close; break;
						case  1: core->state = pm_read_slave; break;
						case -1: core->state = pm_close; break;
						}
					break;
				}
		}
}

int
pm_core_init (struct ev_loop * loop, sfd_t sfd, state_t state, core_t ** core)
{
  if (( * core = malloc (sizeof (core_t))) == NULL)
    {
      pm_log_error ("pm_core_init () - malloc ()", PM_PRINT_ERRNO);
      return -1;
    }

	/** set file descriptor no block */
	fcntl (sfd, F_SETFL, fcntl (sfd, F_GETFL) | O_NONBLOCK);

  ( * core)->sfd = sfd;
  ( * core)->state = state;
  ( * core)->closed = 0;

	//change_curr_slave (& pool);
  
	/** change current slaves */
	pm_slaves_pool_change_curr (& pool);

	( * core)->slave = pool->curr;
	( * core)->rbuffer = NULL;
	( * core)->rbuffer_bytes = 0;

	( * core)->wbuffer = NULL;
	( * core)->wbuffer_bytes = 0;

	( * core)->loop = loop;
	( * core)->event.data = (void * ) ( * core);
	
  return 0;
}

void
pm_core_close (core_t * core)
{
	fprintf (stdout, "close le fd %d\n", core->sfd);
	pm_remove_fd (core->loop, & core->event);
	close (core->sfd);
	if (core->rbuffer != NULL)
		free (core->rbuffer);
	if (core->wbuffer != NULL)
		free (core->wbuffer);
	free (core);
}

int
pm_core_parse_json (char * buffer, core_t ** core)
{
	/** parse json message */
	( * core)->json_obj = json_tokener_parse (buffer);
	if (!( * core)->json_obj)
		{
			pm_log_error ("message format is not json format", PM_NOPRINT_ERRNO);
			return -1;
		}
	return 0;
}

int
pm_core_json_cmd (struct json_object * json_obj, core_t ** core)
{
	struct json_object * json_cmd = NULL;
	struct json_object * json_key = NULL;
	char cmd_string[41];
	
	json_cmd = json_object_object_get (json_obj, PM_MSG_COMMAND);
	if (!json_cmd)
		{
			pm_log_error ("no command in json message", PM_NOPRINT_ERRNO);
			return -1;
		}
	strcpy (cmd_string, json_object_get_string (json_cmd));
	
	if ((strcmp (cmd_string, PM_STR_CMD_ADD) == 0 && (( * core)->cmd = PM_CMD_ADD)) ||
			(strcmp (cmd_string, PM_STR_CMD_SET) == 0 && (( * core)->cmd = PM_CMD_SET)) ||
			(strcmp (cmd_string, PM_STR_CMD_DEL) == 0 && (( * core)->cmd = PM_CMD_DEL)) ||
			(strcmp (cmd_string, PM_STR_CMD_GET) == 0 && (( * core)->cmd = PM_CMD_GET)))
		{
	
			json_key = json_object_object_get (json_obj, PM_MSG_DATA_KEY);
			if (!json_key)
				{
					pm_log_error ("no key in json message", PM_NOPRINT_ERRNO);
					return -1;
				}
			strncpy (( * core)->key, json_object_get_string (json_key), 41);
			
			return 0;
		}
	
	pm_log_error ("command is invalid", PM_NOPRINT_ERRNO);
	return -1;
}
